/**
 * Copyright (c) 2023 murenchao
 * taomu is licensed under Mulan PubL v2.
 * You can use this software according to the terms and conditions of the Mulan PubL v2.
 * You may obtain a copy of Mulan PubL v2 at:
 *       http://license.coscl.org.cn/MulanPubL-2.0
 * THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
 * EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
 * MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
 * See the Mulan PubL v2 for more details.
 */
package cool.taomu.thread

import cool.taomu.entity.DataEntity
import cool.taomu.entity.DataEntity.StateType
import cool.taomu.entity.TimerTaskEntity
import cool.taomu.entity.TimerTaskEntity.Operation
import cool.taomu.lang.AbsWrapperClassLoader
import cool.taomu.lang.GroovyWrapperClassLoader
import cool.taomu.lang.MultiClassLoader.GroovyMultiClassLoader
import cool.taomu.lang.MultiClassLoader.JarMultiClassLoader
import cool.taomu.lang.Runnable
import cool.taomu.lang.TaomuEntry
import cool.taomu.lang.TaomuRuntime
import cool.taomu.utils.ObjectInputStreamAware
import cool.taomu.utils.ThriftUtils
import java.io.InputStream
import java.net.InetSocketAddress
import java.nio.ByteBuffer
import java.util.Arrays
import java.util.UUID
import java.util.concurrent.ConcurrentHashMap
import java.util.concurrent.ExecutorService
import java.util.concurrent.ScheduledFuture
import org.apache.commons.lang3.SerializationException
import org.apache.commons.lang3.SerializationUtils
import org.apache.thrift.TException
import org.apache.thrift.TProcessor
import org.checkerframework.checker.units.qual.min
import org.slf4j.LoggerFactory

class NetworkThreadServer implements NetworkThread.Iface {
	val LOG = LoggerFactory.getLogger(NetworkThreadServer);
	val static cml = JarMultiClassLoader.instance;
	val static gml = GroovyMultiClassLoader.instance;
	var static String host;
	var timerTasks = new ConcurrentHashMap<String, ScheduledFuture<?>>();

	override execute(ByteBuffer data) throws TException {
		try {
			var ser = Arrays.copyOfRange(data.array(), data.position(), data.limit());
			var de = SerializationUtils.deserialize(ser) as DataEntity;
			var key = de.uuid === null ? UUID.randomUUID.toString : de.uuid;
			val runnable = getExecuteRunnable(de, key);
			if (runnable !== null) {
				if (de.timerTask !== null) {
					threadPool(de.timerTask, [
						threadPool(runnable);
					]);
				} else {
					threadPool(runnable);
				}
			}
		} catch (SerializationException ex) {
			LOG.info("反序列化失败", ex);
		} catch (Exception ex) {
			LOG.info("Exception:", ex);
		}
	}

	def threadPool(TimerTaskEntity tte, Runnable runnable) {
		var scheduled = TaomuThreadPoolFactory.getscheduledThreadPool;
		if (!timerTasks.containsKey(tte.name) && tte.operation === Operation.START) {
			LOG.info("启动定时任务:{}", tte.name);
			switch (tte.type) {
				case SCHEDULE: {
					timerTasks.put(tte.name, scheduled.schedule(runnable, tte.delay, tte.timeUit));
				}
				case SCHEDULE_AT_FIXED_RATE: {
					timerTasks.put(tte.name,
						scheduled.scheduleAtFixedRate(runnable, tte.initialDelay, tte.delay, tte.timeUit));
				}
				case SCHEDULE_WITH_FIXED_DELAY: {
					timerTasks.put(tte.name,
						scheduled.scheduleWithFixedDelay(runnable, tte.initialDelay, tte.delay, tte.timeUit));
				}
			}
		} else if (timerTasks.containsKey(tte.name) && tte.operation === Operation.STOP) {
			LOG.info("关闭定时任务:{}", tte.name);
			timerTasks.get(tte.name).cancel(true);
		}
	}

	def threadPool(TaomuRuntime runnable) {
		var a = TaomuThreadPoolFactory.getThreadPool;
		a.execute(runnable)
	}

	def TaomuRuntime getExecuteRunnable(DataEntity de, String key) {
		LOG.info("获取runnable实例:{}", de);
		switch (de.type) {
			case GROOVY: {
				var ret = new TaomuRuntime() {
					override run() {
						synchronized (gml) {
							var script = de.script;
							try {
								if (gml.cache.containsKey(key) && de.stateType == StateType.REDEPLOYMENT) {
									gml.cache.remove(key);
									var g = new GroovyWrapperClassLoader(script, cml);
									gml.cache.put(key, g);
								} else if (!gml.cache.containsKey(key)) {
									var g = new GroovyWrapperClassLoader(script, cml);
									gml.cache.put(key, g);
								}
								var mclass = gml.cache.get(key).loadClass(de.main_class);
								var TaomuEntry te = mclass.newInstance as TaomuEntry;
								te.entry(de.payload);
							} catch (Exception ex) {
								LOG.debug("RuntimeException:",ex);
								NetworkThreadClient.execute(NetworkThreadServer.host,de);
							}
						}
					}
				}
				ret.data = de;
				return ret;
			}
			default: {
				return null;
			}
		}
	}

	def getObjectInputStreamAware(InputStream bais, AbsWrapperClassLoader dest) {
		return new ObjectInputStreamAware(bais, dest);
	}

	def static start(String host, int port, int max, int min) {
		NetworkThreadServer.host = host;
		var TProcessor tprocessor = new NetworkThread.Processor(new NetworkThreadServer());
		var isa = new InetSocketAddress(host, port);
		ThriftUtils.startServer(tprocessor, isa, max, min);
	}
	
	def static start(String host, int port,  ExecutorService pool, int selector) {
		NetworkThreadServer.host = host;
		var TProcessor tprocessor = new NetworkThread.Processor(new NetworkThreadServer());
		var isa = new InetSocketAddress(host, port);
		ThriftUtils.startServer(tprocessor, isa, pool, selector);
	}

}
